package cn.edu.fzu.delivery.helper;

import static cn.edu.fzu.delivery.http.HttpClient.JSON_HEADER;
import static java.util.concurrent.TimeUnit.MILLISECONDS;

import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import cn.edu.fzu.delivery.io.DingDingAt;
import cn.edu.fzu.delivery.io.DingDingMessage;
import cn.edu.fzu.delivery.io.MarkdownMessage;
import cn.edu.fzu.delivery.http.HttpClient;
import cn.edu.fzu.delivery.http.HttpClientFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;

/**
 * 钉钉机器人
 * 
 * <pre>
 *  机器人发送消息频率限制: 每个机器人每分钟最多发送 20 条. 如果超过 20 条, 会限流 10 分钟
 *  
 *  自定义机器人: https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq
 * </pre>
 */
public final class DingDingRobot extends Thread {

    private static final Logger                              log         = LoggerFactory.getLogger(DingDingRobot.class);

    private static final HttpClient                          HC          = HttpClientFactory.get(1000, 2000);

    private static final LoadingCache<String, DingDingRobot> ROBOT_CACHE = CacheBuilder.newBuilder().initialCapacity(1)
            .build(
                    // cache loader
                    new CacheLoader<String, DingDingRobot>() {
                        @Override
                        public DingDingRobot load(String key) {
                            return new DingDingRobot(key);
                        }
                    });

    private final String                                     webhookUrl;
    private final BlockingQueue<Message>                     messageQueue;
    private final BlockingQueue<Message>                     delayQueue;
    private final Counter                                    counter;

    // 返回 [快递] 钉钉群机器人
    public static DingDingRobot delivery() {
        return DingDingRobot.getRobot(
                "https://oapi.dingtalk.com/robot/send?access_token=ac5c40ec38e2831cf078ef367bd156b4d0d2c4d453b602eb4a08255eaeadbe7d");
    }


    private DingDingRobot(final String webhookUrl) {

        this.webhookUrl = webhookUrl;

        // 此计数器不适用于分布式环境
        counter = new Counter();

        // 消息放入队列, 异步发送
        messageQueue = new LinkedBlockingQueue<>();
        // 若一个周期内发送消息数量已达到阈值, 将消息放到此队列以阻塞发送, 直至下一个计数周期再发送
        delayQueue = new DelayQueue<>();

        start();
    }


    /**
     * 返回一个钉钉群机器人, 使用 markdown 格式组织消息
     * 
     * @param webhookUrl 机器人 webhook 地址
     * @return DingDingRobot
     */
    public static final DingDingRobot getRobot(String webhookUrl) {
        return ROBOT_CACHE.getUnchecked(webhookUrl);
    }


    /** 停止所有机器人 **/
    public static final void shutdownAll() {
        Collection<DingDingRobot> robots = ROBOT_CACHE.asMap().values();
        for (DingDingRobot robot : robots) {
            robot.shutdown();
        }
    }


    /**
     * 发消息
     * 
     * @param title 消息标题
     * @param content 消息详情
     * @param items 扩展信息, 可以为 null
     */
    public void send(final String title, final String content, final List<String> items, List<String> mobiles,
                     String userId) {
        messageQueue.add(new Message(title, content, items, mobiles, userId));
    }


    public void shutdown() {
        interrupt();
        log.info("dingding robot shutdown, discard message: {}, {}", messageQueue.size(), delayQueue.size());
    }


    @Override
    public void run() {
        try {
            consume();
        } catch (InterruptedException e) {
            log.warn("consume interrupted");
            interrupt();
        }
    }


    private void consume() throws InterruptedException {
        while (!isInterrupted()) {
            Message message = messageQueue.take();
            if (!counter.increase()) {
                // 若当前计数周期计数器已满, 将消息放到 delay queue, 等到下个计数周期再发送
                message.expireTime = counter.nextStart();
                delayQueue.add(message);
                message = delayQueue.take();
                counter.increase();
            }
            send(message);
        }
        log.warn("interrupted");
    }


    private final void send(final Message message) {
        MarkdownMessage markdown = message.markdown;
        log.debug("send ddmsg: title={}, text={}", markdown.getTitle(), markdown.getText());
        try {
            HC.post(webhookUrl, JSON_HEADER, JSON.toJSONString(new DingDingMessage(markdown, message.at)));
        } catch (Exception e) {
            log.error(e.getMessage());
        }

    }

    private static final class Message implements Delayed {

        private final MarkdownMessage markdown;
        private final long            nanoTime;

        private final DingDingAt      at;
        // 到期时间戳
        private long                  expireTime;

        private Message(String title, String content, List<String> items, List<String> mobiles, String userId) {
            this.nanoTime = System.nanoTime();
            this.markdown = new MarkdownMessage(mobiles, title, content, items.toArray(new String[] {}));
            this.at = new DingDingAt(mobiles, userId);
        }


        @Override
        public int compareTo(Delayed o) {
            return Long.compare(nanoTime, ((Message) o).nanoTime);
        }


        @Override
        public long getDelay(TimeUnit unit) {
            long delay = expireTime - System.currentTimeMillis();
            switch (unit) {
                case MILLISECONDS:
                    return delay;
                case MICROSECONDS:
                    return MILLISECONDS.toMicros(delay);
                case NANOSECONDS:
                    return MILLISECONDS.toNanos(delay);
                default:
                    throw new IllegalArgumentException("unsupport timeunit: " + unit.name());
            }
        }
    }

    private static final class Counter {

        // 一个计数周期 = 60,000 毫秒 
        private static final long CYCLE_DURATION = 60L * 1000L;

        // 一个计数周期内最大吞吐量 = 20
        private static final int  MAX_THROUGHPUT = 20;

        // 本计数周期的开始时间戳
        private long              start;
        // 本计数周期内已发送消息数量
        private int               count;

        private Counter() {
            start = System.currentTimeMillis();
            count = 0;
        }


        /** 尝试将计数器 +1 **/
        private boolean increase() {

            long duration = System.currentTimeMillis() - start;

            // 当前周期且发送数量超过阈值
            if (duration < CYCLE_DURATION && count >= MAX_THROUGHPUT) {
                return false;
            }

            if (duration >= CYCLE_DURATION) {
                // 已经跨周期, 重置
                start = System.currentTimeMillis();
                count = 1;
            } else {
                // 当前周期, 计数器 +1
                count += 1;
            }
            return true;
        }


        /** 获取下一个计数周期开始的时间戳 **/
        private long nextStart() {
            return start + CYCLE_DURATION;
        }
    }

}
